package com.corn.kafkastream.demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Printed;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2022/9/14 15:34
 */
@Slf4j
public class KafkaStreamsSelectKeyApiTest {

    private static final String BOOTSTRAP_SERVER = "node1:9092";

    private static final String INPUT_TOPIC = "stream.input";

    private static final String OUTPUT_TOPIC = "stream.output";

    private static final String APP_ID = "SELECT_KEY_APP_ID";

    public static void main(String[] args) throws InterruptedException {
        // 1. 创建配置
        Properties properties = new Properties();
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID);
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        // 2. 创建streamBuilder
        StreamsBuilder builder = new StreamsBuilder();
        builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), Serdes.String()).withName("input-stream"))
                .peek((k, v) -> {
                    log.info("input value:{}", v);
                })
                .selectKey((k, v) -> {
                    return v.toUpperCase();
                })
                .print(Printed.toSysOut());

        KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties);

        CountDownLatch countDownLatch = new CountDownLatch(1);
        kafkaStreams.start();
        countDownLatch.await();
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            kafkaStreams.close();
            countDownLatch.countDown();
        }));
    }
}
